package utils

import (
	"fmt"
	"runtime/debug"
	"sync"
	"sync/atomic"

	"gddgame.cc/galaxy/utils/def"
	"gddgame.cc/galaxy/utils/hash"
)

type ThreadExec func()

type ThreadExecNo func(no int)

type ThreadNotice func(thread *Threader)

type ThreadExecutor struct {
	fn   ThreadExec
	Time int64
}

type ThreaderManager interface {
	ChanClose() chan struct{}
}

// 确保动态扩容channel
type Threader struct {
	def.ID
	no     int
	logger def.Logger
	safe   bool

	close  chan struct{}
	listen bool
	e      chan ThreadExecutor
	ee     chan struct{}
	notice ThreadNotice

	pool *ThreaderPool

	manager ThreaderManager

	// 动态变更buffer
	buffer        int32
	running       int32
	fnSlice       []*ThreadExecutor
	mutex         sync.Mutex
	inPoolRunning bool
}

func NewThread(id def.ID, buffer int) *Threader {
	threader := &Threader{
		ID:   id,
		safe: false,
		e:    make(chan ThreadExecutor, buffer),
		//ee:      make(chan struct{}, 10),
		buffer:  int32(buffer) - 2,
		running: 0,
		fnSlice: make([]*ThreadExecutor, 0, buffer),
		mutex:   sync.Mutex{},
	}

	return threader
}

func (tm *Threader) Logger(logger def.Logger) {
	tm.logger = logger
}

func (tm *Threader) Notice(notice ThreadNotice) {
	tm.notice = notice
}

func (tm *Threader) Less(other interface{}) bool {
	return tm.running < other.(*Threader).running
}

func (tm *Threader) Equal(other interface{}) bool {
	//fmt.Println(tm, other)
	return tm == other.(*Threader)
}

func (tm *Threader) Thread(fn func()) {
	if tm.safe {
		tm.thread(fn)
	} else if tm.pool != nil {
		tm.pool.Thread(fn)
	} else {
		go fn()
	}
}

func (tm *Threader) thread(fn func()) {
	executor := &ThreadExecutor{
		fn: fn,
		//Time: time.Now().UnixNano() / 1e6,
	}

	t := atomic.AddInt32(&tm.running, 1)
	//if strings.Contains(tm.ID.ID(), "path") {
	//	fmt.Println("thread", tm.running)
	//}
	if t >= tm.buffer {
		atomic.AddInt32(&tm.running, -1)
		tm.mutex.Lock()
		tm.fnSlice = append(tm.fnSlice, executor)
		tm.mutex.Unlock()
	} else {
		// 如果fn通过channel进行同步等待，可能会产生消息阻塞
		// 当进行同步等待，进入到同一个thread,就产生了阻塞
		tm.e <- *executor
	}
}

func (tm *Threader) CleanThread() {
	if tm.listen {
		tm.listen = false
	}
	if tm.close != nil {
		close(tm.close)
		tm.close = nil
	}
	atomic.StoreInt32(&tm.running, 0)
	tm.fnSlice = tm.fnSlice[:]
}
func (tm *Threader) listenThread() {
	if tm.listen {
		return
	}
	tm.close = make(chan struct{}, 1)
	cc := tm.close
	if tm.manager != nil {
		cc = tm.manager.ChanClose()
	}
	go func() {
		tm.listen = true
		defer func() {
			if err := recover(); err != nil {
				debug.PrintStack()
				fmt.Println("thread", err)
				//tm.CleanThread()
				tm.listenThread()
			}
		}()
		for {
			select {
			case e := <-tm.e:
				if e.fn != nil {
					e.fn()
				}
				var first *ThreadExecutor = nil
				tm.mutex.Lock()
				if tm != nil && len(tm.fnSlice) > 0 {
					first = tm.fnSlice[0]
					tm.fnSlice = tm.fnSlice[1:]
				}
				tm.mutex.Unlock()
				//fmt.Println("run3", tm.running)

				if first != nil {
					tm.e <- *first
				} else {
					atomic.AddInt32(&tm.running, -1)
					if tm.notice != nil {
						tm.notice(tm)
					}
				}
				//tm.ee <- struct{}{}
				//fmt.Println("run4", tm.running)
			case <-tm.close:
				tm.listen = false
				return
			case <-cc:
				tm.listen = false
				return
			}
		}
	}()
}

func (tm *Threader) GetThread() bool {
	return tm.safe
}

func (tm *Threader) SetThread(flag bool) {
	tm.safe = flag
	if tm.safe {
		//if tm.logger != nil {
		//	tm.logger.Debugf("Start Thread")
		//}
		tm.listenThread()
	} else if tm.listen {
		//if tm.logger != nil {
		//	tm.logger.Debugf("End Thread")
		//}
		tm.CleanThread()
	}

}

func (tm *Threader) SetThreadPool(pool *ThreaderPool) {
	tm.pool = pool
}

func (tm *Threader) SetManager(manager ThreaderManager) {
	tm.manager = manager
	tm.SetThread(manager != nil)
}

// 保证线程复用（go协程）：大量并发，交由系统进行阻塞
type ThreaderPool struct {
	def.ID
	//logger     def.Logger
	buffer  int
	maxIdle int
	no      int

	close           chan struct{}
	noticeChan      chan *Threader
	activeMutex     sync.Mutex
	activeThreaders []*Threader
	statMutex       sync.Mutex
	idles           []int
	running         []int
	empty           []int
	pool            sync.Pool
}

func NewThreaderPool(id def.ID, buffer int, maxIdle int) *ThreaderPool {
	p := &ThreaderPool{
		ID:              id,
		buffer:          buffer,
		maxIdle:         maxIdle,
		close:           make(chan struct{}, 1),
		noticeChan:      make(chan *Threader, 10),
		activeMutex:     sync.Mutex{},
		activeThreaders: make([]*Threader, 0, 30),
		statMutex:       sync.Mutex{},
		idles:           []int{},
		running:         []int{},
		empty:           []int{},
		pool:            sync.Pool{},
	}
	p.pool.New = func() interface{} {
		t := NewThread(p.ID, p.buffer)
		t.notice = p.notice
		return t
	}
	p.listenNotice()
	return p
}

func (p *ThreaderPool) listenNotice() {
	go func() {
		for {
			select {
			case t := <-p.noticeChan:
				p.idleUpdate(t)
			case <-p.close:
				return
			}
		}
	}()
}

func (p *ThreaderPool) idleUpdate(threader *Threader) {
	if len(p.idles) >= p.maxIdle {
		threader.CleanThread()
		p.activeMutex.Lock()
		p.activeThreaders[threader.no] = nil
		p.empty = append(p.empty, threader.no)
		p.activeMutex.Unlock()
		p.pool.Put(threader)
	} else {
		p.statMutex.Lock()
		p.idles = append(p.idles, threader.no)
		p.statMutex.Unlock()
	}
}

func (p *ThreaderPool) notice(threader *Threader) {
	//fmt.Println("notice", "fff", threader.running)
	if threader.running == 0 {
		p.noticeChan <- threader
	}
}
func (p *ThreaderPool) Thread(fn ThreadExec) {
	var t *Threader = nil
	var index int = -1
	var in = false

	// 查找空闲线程
	// 		加入运行线程
	// 查找运行线程
	// 		判断线程是否还存在，循环读取 (超过最大空闲，则已经释放)
	// 		判断线程运行数（大于当前运行线程数），移除运行线程
	// 创建新线程
	// 		判断空的线程位，存在则占用
	//		加入新的线程位 （线程保持线程引用）
	// 线程运行完
	//		判断是否运行数为空
	//		加入空闲线程
	// 		如果空闲线程数量超过最大设定
	// 		停止线程，加入缓存，治空线程位
	//fmt.Println("pool", p.ID, 1)

	for {
		p.statMutex.Lock()
		ld := len(p.idles)
		if ld > 0 {
			index = p.idles[0]
			p.idles = p.idles[1:]
		}
		p.statMutex.Unlock()
		//fmt.Println("pool", p.ID, 2, index)
		if index >= 0 {
			t = p.activeThreaders[index]
			if t == nil {
				index = -1
				continue
			}
			//fmt.Println("pool", index, p.activeThreaders, p.running, p.idles, p.empty)
			if t.inPoolRunning {
				in = false
			}
			break
		} else {
			break
		}
	}

	if t == nil {
		for {
			p.statMutex.Lock()
			ll := len(p.running)
			if ll > 0 {
				index = p.running[0]
				p.running = p.running[1:]
			}
			p.statMutex.Unlock()
			if index >= 0 {
				t = p.activeThreaders[index]
				if t == nil {
					index = -1
					continue
				}
				if t.running > int32(p.buffer/2) {
					t.inPoolRunning = false
				}
				in = t.inPoolRunning
				break
			} else {
				break
			}
		}
	}
	//fmt.Println("pool", p.ID, 1)

	if t == nil {
		t = p.pool.Get().(*Threader)
		t.notice = p.notice
		t.SetManager(p)

		p.activeMutex.Lock()
		index := p.no
		if len(p.empty) > 0 {
			index = p.empty[0]
			p.empty = p.empty[1:]
			p.activeThreaders[index] = t
		} else {
			p.no += 1
			p.activeThreaders = append(p.activeThreaders, t)
		}
		p.activeMutex.Unlock()

		t.no = index
		t.inPoolRunning = true
		in = t.inPoolRunning
	}

	if in {
		p.statMutex.Lock()
		p.running = append(p.running, t.no)
		p.statMutex.Unlock()
	}
	t.thread(fn)
}

func (p *ThreaderPool) Clean() {
	close(p.close)
	p.close = make(chan struct{}, 1)
	p.activeThreaders = p.activeThreaders[0:0]
	// todo 放入pool
}

func (p *ThreaderPool) ReStart() {
	p.listenNotice()
}

func (p *ThreaderPool) ChanClose() chan struct{} {
	return p.close
}

// 保证固定线程数
type ThreaderBucket struct {
	def.ID
	//logger     def.Logger
	bucket int
	buffer int

	close           chan struct{}
	activeThreaders map[string]*Threader
	threaderMutex   sync.RWMutex

	*hash.Consistent
}

func NewThreaderBucket(id def.ID, number int, buffer int) *ThreaderBucket {
	p := &ThreaderBucket{
		ID:              id,
		bucket:          number,
		buffer:          buffer,
		close:           make(chan struct{}, 1),
		activeThreaders: make(map[string]*Threader, number),
		Consistent:      hash.NewConsistent(),
	}
	s := make([]string, number)
	for i := 0; i < number; i++ {
		s[i] = ToString(i)
	}
	p.Consistent.Set(s)
	return p
}

func (p *ThreaderBucket) Thread(key string, fn ThreadExecNo) {
	n, _ := p.Get(key)
	p.threaderMutex.RLock()
	t, ok := p.activeThreaders[n]
	p.threaderMutex.RUnlock()
	if !ok {
		p.threaderMutex.Lock()
		t, ok = p.activeThreaders[n]
		if !ok {
			t = NewThread(p.ID, p.buffer)
			t.no = ToInt(n)
			t.SetManager(p)
			p.activeThreaders[n] = t
		}
		p.threaderMutex.Unlock()
	}
	no := t.no
	t.Thread(func() {
		fn(no)
	})
}

func (p *ThreaderBucket) Clean() {
	close(p.close)
	p.close = make(chan struct{}, 1)
}

func (p *ThreaderBucket) ChanClose() chan struct{} {
	return p.close
}
